package indicator;

import com.alibaba.druid.util.StringUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.wushi.redis.RedisUtil;
import jdk.internal.dynalink.beans.StaticClass;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import redis.clients.jedis.JedisPool;
import utils.DateUtil;

import java.util.*;

import static indicator.IndicatorCommonUVService.*;


/**
 * @ClassName indicator.CommonUVBroadCastProcessFunction
 * @Deacription TODO
 * @Author wushumin
 * @Date 2021-06-23 16:42
 * @Version 1.0
 **/
@Slf4j
public class CommonUVBroadCastProcessFunction extends KeyedBroadcastProcessFunction<String, JSONObject, Map<Long, IndicatorConfig>, String> {
    private static final String COMMONUV_CACHE_KEY = "mars:indicator:mic:commonuv:%s";
    private static JedisPool jedisPool = com.wushi.redis.RedisUtil.getJedisPool("mars-redis.lcbint.cn");

    /**
     * 节后指标配置数据，处理UV
     *
     * @param event
     * @param ctx
     * @param out
     * @throws Exception
     */
    @Override
    public void processElement(JSONObject event, ReadOnlyContext ctx, Collector<String> out) throws Exception {
        log.info("======>process indicator config data:{}", JSON.toJSONString(event));
        List<String> updatedIndicators = new ArrayList<>();
        ReadOnlyBroadcastState<String, Map<Long, IndicatorConfig>> broadcastState = ctx.getBroadcastState(INDICATOR_STATE_DESC);
        IndicatorConfig indicator = null;
        if (broadcastState.contains(INDICATOR_CONFIG_STATE)) {
            Map<Long, IndicatorConfig> indicatorConfigState = broadcastState.get(INDICATOR_CONFIG_STATE);
            if (indicatorConfigState == null || indicatorConfigState.size() <= 0) {
                log.error("======>the indicatorConfig is null");
                return;
            }
            for (Map.Entry<Long, IndicatorConfig> entry : indicatorConfigState.entrySet()) {
                IndicatorConfig indicatorConfig = entry.getValue();
                String tmp = doProcessUV(indicatorConfig, event);
                if (tmp != null) {
                    updatedIndicators.add(tmp);
                }
            }
        }
        if (updatedIndicators.size() > 0) {
            out.collect(event.getString(USER_ID) + JSON.toJSONString(updatedIndicators));
        }
    }

    private String getUVHashKey() {
        return String.format(COMMONUV_CACHE_KEY, DateUtil.getCurrentDateStr());
    }

    /**
     * 具体处理UV，在redis中保存userId
     *
     * @param indicatorConfig
     * @param event
     * @return
     */
    private String doProcessUV(IndicatorConfig indicatorConfig, JSONObject event) {
        String dimenstions = indicatorConfig.getDimenstions();
        JSONObject jsonObject = JSON.parseObject(dimenstions);
        StringBuffer dimentionKeys = new StringBuffer();
        for (String dimentionKey : jsonObject.keySet()) {
            DimenstionEnum dimenstionEnum = DimenstionEnum.ofDimensionKey(dimentionKey);
            if (dimenstionEnum == null) {
                continue;
            }
            if (!event.containsKey(dimenstionEnum.getDimensionKeyInEvent())) {
                return null;
            }
            dimentionKeys.append(event.getString(dimenstionEnum.getDimensionKeyInEvent())).append("_");
        }
        String uvKey = indicatorConfig.getIndicatorCode() + ":" + dimentionKeys.deleteCharAt(dimentionKeys.length() - 1).toString();
        //cache:   key-groupId_pageId_activitID
        String commonUVhashkey = getUVHashKey();
        List<String> userIdDatas = RedisUtil.hashGetValues(commonUVhashkey, uvKey);
        Long isNewUser = null;
        if (CollectionUtils.isEmpty(userIdDatas)) {
            isNewUser = RedisUtil.hashSetValues(commonUVhashkey, uvKey, Arrays.asList(event.getString(USER_ID)), 3600 * 48);
        } else {
            if (!userIdDatas.contains(event.getString(USER_ID))) {
                userIdDatas.add(event.getString(USER_ID));
                isNewUser = RedisUtil.hashSetValues(commonUVhashkey, uvKey, userIdDatas, 3600 * 48);
            } else {
                return null;
            }
        }
        log.debug("======>finished indicator:{},event:{}, newUser:{}", dimenstions, event, isNewUser);
        return commonUVhashkey + "_" + uvKey;
    }

    /**
     * 更新 IndicatorConfig 的state
     *
     * @param values
     * @param ctx
     * @param out
     * @throws Exception
     */
    @Override
    public void processBroadcastElement(Map<Long, IndicatorConfig> values, Context ctx, Collector<String> out) throws Exception {
        BroadcastState<String, Map<Long, IndicatorConfig>> broadcastState = ctx.getBroadcastState(INDICATOR_STATE_DESC);
        log.info("======>process broadcast data:{}", JSON.toJSONString(values));
        broadcastState.put(INDICATOR_CONFIG_STATE, values);
    }
}
